#pragma once
#include "muduo/proto/codec.h"
#include "muduo/net/TcpConnection.h"

#include "../mqcommon/helper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/request.pb.h"
#include "consumer.hpp"
#include <mutex>
#include <condition_variable>

using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
using basicConsumerResponsePtr = std::shared_ptr<basicConsumerResponse>;
using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;

class Channel
{
public:
    using Ptr = std::shared_ptr<Channel>;
    Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        :_cid(UUIDHelper::uuid()), _conn(conn), _codec(codec)
    {}
    std::string getCid()
    {
        return _cid;
    }
    ~Channel() 
    {
        DLOG("信道关闭");
        basicCancel();
        DLOG("信道关闭");
    }
    bool openChannel()
    {
       // DLOG("进来了吗，到哪了");
        openChannelRequest req;
       // DLOG("进来了吗，到哪了");
        req.set_cid(_cid);
       // DLOG("进来了吗，到哪了");
        req.set_rid(UUIDHelper::uuid());
      //  DLOG("进来了吗，到哪了");
      //  DLOG("%p", _codec.get());
      //  DLOG("%p", _conn.get());
        _codec->send(_conn, req);
      //  DLOG("进来了吗，到哪了");
        basicCommonResponsePtr resp = waitResponse(req.rid());
      //  DLOG("进来了吗，到哪了");
        return resp->ok();
    }
    void closeChannel()
    {
        closeChannelRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());

        _codec->send(_conn, req);
        waitResponse(req.rid());
    }
    bool declareExchange(const std::string& exchangeName, ExchangeType type, bool durable, bool autoDelete, google::protobuf::Map<std::string, std::string>& args)
    {
        declareExchangeRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_exchange_name(exchangeName);
        req.set_exchange_type(type);
        req.set_durable(durable);
        req.set_auto_delete(autoDelete);
        req.mutable_args()->swap(args);

        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(req.rid());
        return resp->ok();
    }
    void deleteExchange(const std::string& exchangeName)
    {
        deleteExchangeRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_exchange_name(exchangeName);

        _codec->send(_conn, req);
        waitResponse(req.rid());
    }

    bool declareMsgQueue(const std::string& msgQueueName, bool exclusive, bool durable, bool autoDelete, google::protobuf::Map<std::string, std::string>& args)
    {
        declareMsgQueueRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_msgqueue_name(msgQueueName);
        req.set_exclusive(exclusive);
        req.set_durable(durable);
        req.set_auto_delete(autoDelete);
        req.mutable_args()->swap(args);
        // DLOG("durable:%d不对吗", durable);////////////////////////////////////////////
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(req.rid());
        return resp->ok();
    }
    void deleteMsgQueue(const std::string& msgQueueName)
    {
        deleteMsgQueueRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_msgqueue_name(msgQueueName);

        _codec->send(_conn, req);
        waitResponse(req.rid());
    }

    bool bind(const std::string& exchangeName, const std::string& msgQueueName, const std::string& bindKey)
    {
        bindRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_exchange_name(exchangeName);
        req.set_msgqueue_name(msgQueueName);
        req.set_binding_key(bindKey);

        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(req.rid());
        return resp->ok();
    }
    void unBind(const std::string& exchangeName, const std::string& msgQueueName)
    {
        unBindRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_exchange_name(exchangeName);
        req.set_msgqueue_name(msgQueueName);

        _codec->send(_conn, req);
        waitResponse(req.rid());
    }

    void basicPublish(const std::string& exchangeName, const BasicProperty* property, const std::string& body)//为什么没有返回值
    {
        basicPublishRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_exchange_name(exchangeName);
        if(property != nullptr)
        {
            req.mutable_bp()->set_id(property->id());
            req.mutable_bp()->set_delivery_mode(property->delivery_mode());
            req.mutable_bp()->set_routing_key(property->routing_key());
        }
        // else req.mutable_bp()->set_delivery_mode(DeliveryMode::DURABLE); //这个没必要，队列持久化，消息就会持久化
        req.set_body(body);

        _codec->send(_conn, req);
        waitResponse(req.rid());     
    }
    void basicAck(const std::string& msgId)
    {
        if(_consumer.get() == nullptr)      //没有消费者，不完
        {
            DLOG("没有消费者，无法响应");
            return;
        }
        basicAckRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_msgqueue_name(_consumer->queueName);
        req.set_msgid(msgId);

        _codec->send(_conn, req);
        waitResponse(req.rid());   
    }

    bool basicConsumer(const std::string& consumerTag, const std::string& msgQueueName, bool ack, const ConsumerCallback& cb)
    {
        if(_consumer.get() != nullptr)
        {
            DLOG("信道%s,已经存在消费者:%s", _cid.c_str(), _consumer->tag.c_str());           
            return false;
        }
        basicConsumerRequest req;
        req.set_cid(_cid);
        DLOG("%s", _cid.c_str())
        req.set_rid(UUIDHelper::uuid());
        req.set_consumer_tag(consumerTag);
        req.set_msgqueue_name(msgQueueName);
        req.set_auto_ack(ack);

        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(req.rid());
        if(resp->ok() == false)
        {
            DLOG("添加订阅失败");
            return false;
        }
        _consumer = std::make_shared<Consumer>(consumerTag, msgQueueName, ack, cb);
        return pullOnce(msgQueueName);
    }
    bool pullOnce(const std::string& msgQueueName)
    {
        pullOnceRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_msgqueue_name(msgQueueName);
        _codec->send(_conn, req);

        basicCommonResponsePtr resp = waitResponse(req.rid());
        if(resp->ok() == false)
        {
            DLOG("拉取失败,");
            return false;
        }
        return true;
    }
    void basicCancel()
    {
        if(_consumer.get() == nullptr)      //没有消费者，不完
            return;
        basicCancelRequest req;
        req.set_cid(_cid);
        req.set_rid(UUIDHelper::uuid());
        req.set_consumer_tag(_consumer->tag);
        req.set_msgqueue_name(_consumer->queueName);
        DLOG("server调用我了吗");
        _codec->send(_conn, req);
        waitResponse(req.rid());   
        _consumer.reset();                  //取消这个消费者
    }
    void putBasicResponse(const basicCommonResponsePtr& resp)           //提供给链接对象
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _basicResponse.insert({resp->rid(), resp});
        _cv.notify_all(); //one就够了，值获得一个响应
    }
    bool consumer(const basicConsumerResponsePtr& resp)           //给连接对象提供
    {
        if(_consumer.get() == nullptr)
        {
            DLOG("没有订阅消费者, 无法消费消息");
            return false;
        }
        if(_consumer->tag != resp->consumer_tag())
        {
            DLOG("响应的消费者id和这个信道%s的消费者id不一样, 处理消息失败", _cid.c_str());
            return false;
        }
        _consumer->callback(resp->consumer_tag(), resp->mutable_bp(), resp->body());
        return true;
    }
private:
    basicCommonResponsePtr waitResponse(const std::string& rid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait(lock, [&rid, this](){ return _basicResponse.find(rid) != _basicResponse.end();});
        basicCommonResponsePtr resp = _basicResponse[rid];
        _basicResponse.erase(rid);          //拿到了，就删掉
        return resp;
    }

private:
    std::string _cid;
    muduo::net::TcpConnectionPtr _conn;
    ProtobufCodecPtr _codec;
    Consumer::Ptr _consumer;
    std::mutex _mutex;
    std::condition_variable _cv;
    std::unordered_map<std::string, basicCommonResponsePtr> _basicResponse;     //请求od，与响应的映射
};

class ChannelManager
{
public:
    using Ptr = std::shared_ptr<ChannelManager>;
    ChannelManager(){}
    Channel::Ptr openChannel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
    {
        
        std::unique_lock<std::mutex> lock(_mutex);

        Channel::Ptr channelPtr = std::make_shared<Channel>(conn, codec);
        std::cout << channelPtr.get() << std::endl;
        _channels.insert({channelPtr->getCid(), channelPtr});
        return channelPtr;
    }
    void closeChannel(const std::string& cid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _channels.erase(cid);
    }
    Channel::Ptr getChannel(const std::string& cid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(cid);
        if(it == nullptr)
        {
            DLOG("这个信道id:%s,没有对应的信道", cid.c_str());
            return Channel::Ptr();
        }
        return it->second;
    }
private:
    std::mutex _mutex;
    std::unordered_map<std::string, Channel::Ptr> _channels;
};